﻿//--------------------------------------------------------------------------
// 
//  Copyright (c) Microsoft Corporation.  All rights reserved. 
// 
//  File: BlockingCollectionExtensions.cs
//
//--------------------------------------------------------------------------

using System.Collections.Generic;
using System.Threading;

namespace System.Collections.Concurrent
{
    /// <summary>Extension methods for BlockingCollection.</summary>
    public static class BlockingCollectionExtensions
    {
        /// <summary>Adds the contents of an enumerable to the BlockingCollection.</summary>
        /// <typeparam name="T">Specifies the type of the elements in the collection.</typeparam>
        /// <param name="target">The target BlockingCollection to be augmented.</param>
        /// <param name="source">The source enumerable containing the data to be added.</param>
        /// <param name="completeAddingWhenDone">
        /// Whether to mark the target BlockingCollection as complete for adding when 
        /// all elements of the source enumerable have been transfered.
        /// </param>
        public static void AddFromEnumerable<T>(this BlockingCollection<T> target, IEnumerable<T> source, bool completeAddingWhenDone)
        {
            try { foreach (var item in source) target.Add(item); }
            finally { if (completeAddingWhenDone) target.CompleteAdding(); }
        }

        /// <summary>Adds the contents of an observable to the BlockingCollection.</summary>
        /// <typeparam name="T">Specifies the type of the elements in the collection.</typeparam>
        /// <param name="target">The target BlockingCollection to be augmented.</param>
        /// <param name="source">The source observable containing the data to be added.</param>
        /// <param name="completeAddingWhenDone">
        /// Whether to mark the target BlockingCollection as complete for adding when 
        /// all elements of the source observable have been transfered.
        /// </param>
        /// <returns>An IDisposable that may be used to cancel the transfer.</returns>
        public static IDisposable AddFromObservable<T>(this BlockingCollection<T> target, IObservable<T> source, bool completeAddingWhenDone)
        {
            if (target == null) throw new ArgumentNullException("target");
            if (source == null) throw new ArgumentNullException("source");
            return source.Subscribe(new DelegateBasedObserver<T>
            (
                onNext: item => target.Add(item),
                onError: error => { if (completeAddingWhenDone) target.CompleteAdding(); },
                onCompleted: () => { if (completeAddingWhenDone) target.CompleteAdding(); }
            ));
        }

        /// <summary>Creates an IProducerConsumerCollection-facade for a BlockingCollection instance.</summary>
        /// <typeparam name="T">Specifies the type of the elements in the collection.</typeparam>
        /// <param name="collection">The BlockingCollection.</param>
        /// <returns>
        /// An IProducerConsumerCollection that wraps the provided BlockingCollection.
        /// </returns>
        public static IProducerConsumerCollection<T> ToProducerConsumerCollection<T>(
            this BlockingCollection<T> collection)
        {
            return ToProducerConsumerCollection(collection, Timeout.Infinite);
        }

        /// <summary>Creates an IProducerConsumerCollection-facade for a BlockingCollection instance.</summary>
        /// <typeparam name="T">Specifies the type of the elements in the collection.</typeparam>
        /// <param name="collection">The BlockingCollection.</param>
        /// <param name="millisecondsTimeout">-1 for infinite blocking add and take operations. 0 for non-blocking, 1 or greater for blocking with timeout.</param>
        /// <returns>An IProducerConsumerCollection that wraps the provided BlockingCollection.</returns>
        public static IProducerConsumerCollection<T> ToProducerConsumerCollection<T>(
            this BlockingCollection<T> collection, int millisecondsTimeout)
        {
            return new ProducerConsumerWrapper<T>(collection, millisecondsTimeout, new CancellationToken());
        }

        /// <summary>Creates an IProducerConsumerCollection-facade for a BlockingCollection instance.</summary>
        /// <typeparam name="T">Specifies the type of the elements in the collection.</typeparam>
        /// <param name="collection">The BlockingCollection.</param>
        /// <param name="millisecondsTimeout">-1 for infinite blocking add and take operations. 0 for non-blocking, 1 or greater for blocking with timeout.</param>
        /// <param name="cancellationToken">The CancellationToken to use for any blocking operations.</param>
        /// <returns>An IProducerConsumerCollection that wraps the provided BlockingCollection.</returns>
        public static IProducerConsumerCollection<T> ToProducerConsumerCollection<T>(
            this BlockingCollection<T> collection, int millisecondsTimeout, CancellationToken cancellationToken)
        {
            return new ProducerConsumerWrapper<T>(collection, millisecondsTimeout, cancellationToken);
        }

        internal sealed class ProducerConsumerWrapper<T> : IProducerConsumerCollection<T>
        {
            private readonly BlockingCollection<T> _collection;
            private readonly int _millisecondsTimeout;
            private readonly CancellationToken _cancellationToken;

            public ProducerConsumerWrapper(
                BlockingCollection<T> collection, int millisecondsTimeout, CancellationToken cancellationToken) 
            { 
                if (collection == null) throw new ArgumentNullException("bc");
                if (millisecondsTimeout < -1) throw new ArgumentOutOfRangeException("millisecondsTimeout");
                _collection = collection;
                _millisecondsTimeout = millisecondsTimeout;
                _cancellationToken = cancellationToken;
            }

            public void CopyTo(T[] array, int index)
            {
                _collection.CopyTo(array, index);
            }

            public T[] ToArray()
            {
                return _collection.ToArray();
            }

            public bool TryAdd(T item)
            {
                return _collection.TryAdd(item, _millisecondsTimeout, _cancellationToken);
            }

            public bool TryTake(out T item)
            {
                return _collection.TryTake(out item, _millisecondsTimeout, _cancellationToken);
            }

            public IEnumerator<T> GetEnumerator()
            {
                return ((IEnumerable<T>)_collection).GetEnumerator();
            }

            IEnumerator IEnumerable.GetEnumerator()
            {
                return GetEnumerator();
            }

            public void CopyTo(Array array, int index)
            {
                ((ICollection)_collection).CopyTo(array, index);
            }

            public int Count
            {
                get { return _collection.Count; }
            }

            public bool IsSynchronized
            {
                get { return ((ICollection)_collection).IsSynchronized; }
            }

            public object SyncRoot
            {
                get { return ((ICollection)_collection).SyncRoot; }
            }
        }
    }
}